package com.hc.concurrent_likes.event;

import com.alibaba.fastjson.JSONObject;
import com.hc.concurrent_likes.entity.Event;
import com.hc.concurrent_likes.entity.MoodDTO;
import com.hc.concurrent_likes.utils.ConcurrentLikesConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.ThreadPoolExecutor;

@Component
public class MoodConsumer implements ConcurrentLikesConstant {

    private static final Logger logger = LoggerFactory.getLogger(MoodConsumer.class);

    @Autowired
    @Qualifier("redisTemplate")
    private RedisTemplate redisTemplate;

    //单例模式。整个项目中的所有线程都只能创建一个对象
    private static ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {

        if(MoodConsumer.threadPoolExecutor == null){
            synchronized (threadPoolExecutor){
                if(MoodConsumer.threadPoolExecutor == null) {
                    MoodConsumer.threadPoolExecutor = threadPoolExecutor;
                }
            }
        }
    }

    @KafkaListener(topics = {TOPIC_LIKES})
    public void handleLikesEvent(ConsumerRecord record, Acknowledgment acknowledgment){

        //判断消费者获取的数据是否为空
        if (record == null || record.value() == null){
            logger.error("实体的消息的内容为空！");
            return;
        }

        //将消费者获取到的事件由json转化成java对象的方式
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);

        if (event == null){
            logger.error("实体的消息格式错误！");
            return;
        }

        consumerMoodEventByThreadPool(event);

        acknowledgment.acknowledge();
    }

    //消费说说事件
    private void consumerMoodEventByThreadPool(Event event){
        threadPoolExecutor.execute(new MoodTask(event));
        System.out.println("使用线程池的方式...");
    }

    //说说任务
    private class MoodTask implements Runnable{
        private Event event;
        public MoodTask(Event event){
            this.event = event;
        }
        @Override
        public void run() {
            MoodDTO moodDTO = event.getMoodDTO();

            //采用分布式事务的方式
            redisTemplate.execute(new SessionCallback(){
                @Override
                public Object execute(RedisOperations operations) throws DataAccessException {
                    //开启事务
                    operations.multi();

                    //存放到set集合中，key为PRAISE_HASH_KEY，value为moodId
                    redisTemplate.opsForSet().add(PRAISE_HASH_KEY, moodDTO.getId());

                    //存放到set集合中，key为moodId，value为userId
                    //多个用户给一条说说点赞
                    redisTemplate.opsForSet().add(moodDTO.getId(), moodDTO.getUserId());
                    //点赞数+1（模拟直播时用户可不停地点赞）
//                redisTemplate.opsForValue().set(PRAISE_COUNT_KEY, ++count);
                    return operations.exec();//执行事务并提交
                }
            });
        }
    }
}
